package org.ala.tiktools.rpc.plugin.node;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rpc.config.RpcConfigInfo;
import org.ala.tiktools.rpc.exception.RpcLoginException;
import org.ala.tiktools.rpc.msg.ack.RpcAckBody;
import org.ala.tiktools.rpc.msg.login.RpcLoginBody;
import org.ala.tiktools.rpc.msg.login.RpcLoginMsg;
import org.ala.tiktools.rpc.msg.push.RpcPushBody;
import org.ala.tiktools.rpc.msg.push.RpcPushMsg;
import org.ala.tiktools.rpc.netty.attr.RpcAttribute;
import org.ala.tiktools.rpc.plugin.serializer.SerializeThreadPool;
import org.ala.tiktools.rpc.plugin.serializer.kryo.KryoThreadSerializer;
import org.ala.tiktools.tools.id.TimestampIdGenerator;
import org.slf4j.Logger;

import io.netty.channel.Channel;

/**
 * 节点信息
 *
 * @author ala
 * @date 2025-04-07 11:35
 */
public abstract class AbstractNode implements INode {


    static Logger log = LoggerFactory.getLogger("rpc", "node");


    /**
     *  组信息
     *  <p> AlaRpcClient注解中的name，或hosts。负载均衡时的分组（优先name）
     */
    protected String group;
    /**
     *  节点信息
     */
    protected NodeKey nodeKey;
    /**
     *  当前状态
     */
    protected NodeStatusEnums status;
    /**
     *  id生成器
     */
    protected TimestampIdGenerator idGenerator;
    /**
     *  表示链接的channel对象
     */
    protected Channel channel;
    
    
    /**
     * 节点类型标志
     * <p>	tcp,kcp,dubbo,grpc,http，子类自报自己的通讯方式
     */
    protected abstract String type();
    

    public AbstractNode(String group, NodeKey nodeKey) {
        this.nodeKey = nodeKey;
        this.group = group;
        this.status = NodeStatusEnums.NEW;
        //  雪花码长63位，因为一个生成器只给一个node使用，机器码只占1位，那剩下21位给自增码使用
//        idGenerator = SnowIdGenerator.build(41, 1, 21, true);
        //  32位时间戳 + 32位随机数（全程无锁，同一毫秒内同一channel有2^32-1个id，够你并发了，且id一定要有序吗）
        this.idGenerator = TimestampIdGenerator.build();
    }


    /**
     * 状态变化监听
     */
    protected Map<NodeStatusEnums, List<Runnable>> statusListener = new HashMap<>();
    @Override
    public void addStatusListener(NodeStatusEnums status, Runnable run) {
        statusListener.computeIfAbsent(status, _k -> new ArrayList<>()).add(run);
    }
    protected void setStatus(NodeStatusEnums status) {
        //  已经是摘除节点不能做任何状态转换
        if (NodeStatusEnums.KILL.equals(this.status)) {return;}
        this.status = status;
        runStatus();
    }
    /**
     * 执行状态变化监听
     */
    protected void runStatus() {
        List<Runnable> runs =statusListener.get(status);
        if (runs != null) {
            for (Runnable run : runs) {
                try {
                    run.run();
                } catch (Exception e) {
                    log.error("[{} runStatus] 执行状态监听失败 status:{} node:{}", this.getClass().getSimpleName(), status, this.key(), e);
                }
            }
        }
    }


    @Override
    public CompletableFuture<NodeStatusEnums> active() {
        if (NodeStatusEnums.KILL.equals(status)) {
            log.info("[{} active] 已经摘除的节点无法重新激活 node:{}", this.getClass().getSimpleName(), key());
            CompletableFuture<NodeStatusEnums> cf = new CompletableFuture<>();
            cf.complete(NodeStatusEnums.KILL);
            return cf;
        }

        log.info("[{} active] 节点开始激活 node:{}", this.getClass().getSimpleName(), key());
        setStatus(NodeStatusEnums.BEGIN_ACTIVATE);

        //  如果channel不为空，将连接断掉，重连（做法比较暴力，单可处理假死）
        if (channel != null) {
            close();
            channel = null;
        }

        CompletableFuture<Channel> cf = connect(null, c -> {
            setStatus(NodeStatusEnums.UNAVAILABLE);
        });
        CompletableFuture<NodeStatusEnums> rtn = new CompletableFuture<NodeStatusEnums>();
        cf.whenCompleteAsync((c, e) -> {
            //  如果链接成功，发送登录消息
            if (c != null) {
                channel = c;
                log.info("[{} active] 节点连接成功，发送登录消息 node:{}", this.getClass().getSimpleName(), key());

                String username = RpcConfigInfo.INSTANCE.getAuth().getUsername(),
                        password = RpcConfigInfo.INSTANCE.getAuth().getPassword();
                //  发送登录消息
                login(username, password).whenCompleteAsync((b, ex) -> {
                    if (b != null && b) {
                        log.info("[{} active] 节点已激活 channel:{}", this.getClass().getSimpleName(), channel);
                        setStatus(NodeStatusEnums.ACTIVATE);
                        rtn.complete(NodeStatusEnums.ACTIVATE);
                    } else {
                        //  如果是登录失败，后面再登还是会失败，连接直接断掉
                        if ((ex instanceof RpcLoginException)) {
                            log.error(String.format("[%s active] 登录远程主机失败，账号密码错误 channel:%s username:%s password:%s", this.getClass().getSimpleName(), channel, username, password), ex);
                            setStatus(NodeStatusEnums.UNAVAILABLE);
                            rtn.completeExceptionally(ex);
                            close();
                        } else {
                            log.error(String.format("[%s active] 登录异常", this.getClass().getSimpleName()), ex);
                            setStatus(NodeStatusEnums.UNAVAILABLE);
                            rtn.completeExceptionally(ex);
                            close();
                        }
                    }
                });
            } else if (e != null) {
            	log.error("[{} active] 节点连接失败 node:{}", this.getClass().getSimpleName(), key(), e);
            }
        });
        return rtn;
    }


    /**
     * 建立链接
     */
    protected abstract CompletableFuture<Channel> connect(Consumer<Channel> registered, Consumer<Channel> unRegistered);


    @Override
    public CompletableFuture<RpcAckBody> send(String url, Object[] args, Map<String, Object> metadata) {
        CompletableFuture<RpcAckBody> cf = new CompletableFuture<>();

        RpcPushMsg msg = RpcPushMsg.build();
        long id = id();
		msg.getHead().id(id);
        msg.body(RpcPushBody.build().url(url).args(args).metadata(metadata));
        //	主线程序列化好
        if (RpcConfigInfo.INSTANCE.isSerializeInPlace()) {
            byte[] bytes = KryoThreadSerializer.INSTANCE.serialize(msg.getBody());
            channel.eventLoop().execute(() -> {
        		channel.attr(RpcAttribute.ackMapper).get().put(id, cf);
            	channel.writeAndFlush(msg.bodyBytes(bytes));
            });
        }
        //	序列化好，丢给netty
        else {
        	SerializeThreadPool.INSTANCE.serialize(msg.getBody(), bytes -> channel.eventLoop().execute(() -> {
	        	//	在eventLoop线程里面 put hash，全程无锁
	    		channel.attr(RpcAttribute.ackMapper).get().put(id, cf);
	        	channel.writeAndFlush(msg.bodyBytes(bytes));
	    	}));
        }
        return cf;
    }

    @Override
    public void sendVoid(String url, Object[] args) {
        long id = id();
        RpcPushMsg msg = RpcPushMsg.build();
        msg.getHead().id(id);
        msg.body(RpcPushBody.build().url(url).args(args));
        //	主线程序列化好
        if (RpcConfigInfo.INSTANCE.isSerializeInPlace()) {
            byte[] bytes = KryoThreadSerializer.INSTANCE.serialize(msg.getBody());
            channel.writeAndFlush(msg.bodyBytes(bytes));
        }
        //	序列化好，丢给netty
        else {
        	SerializeThreadPool.INSTANCE.serialize(msg.getBody(), bytes -> {
        		channel.writeAndFlush(msg.bodyBytes(bytes));
        	});
        }
    }

    /**
     * 发送登录消息
     * <p>  只有登录成功的连接，才判定为合法链接
     */
    protected CompletableFuture<Boolean> login(String username, String password) {
        CompletableFuture<Boolean> cf = new CompletableFuture<>();

        RpcLoginMsg msg = RpcLoginMsg.build();
    	long id = id();
    	msg.getHead().id(id);
        msg.body(RpcLoginBody.build().username(username).password(password));
        //	主线程序列化好
        if (RpcConfigInfo.INSTANCE.isSerializeInPlace()) {
        	byte[] bytes = KryoThreadSerializer.INSTANCE.serialize(msg.getBody());
            channel.eventLoop().execute(() -> {
            	channel.attr(RpcAttribute.ackMapper).get().put(id, cf);
            	channel.writeAndFlush(msg.bodyBytes(bytes));
            });
        } 
        //	序列化好，然后丢给netty
        else {
          SerializeThreadPool.INSTANCE.serialize(msg.getBody(), bytes -> channel.eventLoop().execute(() -> {
	      		channel.attr(RpcAttribute.ackMapper).get().put(id, cf);
	          	channel.writeAndFlush(msg.bodyBytes(bytes));
	      }));
        }
        return cf;
    }


    /**
     * 强制关闭，不进行状态转换
     * <p>  在节点被摘掉时候用的
     */
    @Override
    public CompletableFuture<Boolean> close() {
        CompletableFuture<Boolean> cf = new CompletableFuture<Boolean>();
        if (channel != null && channel.isActive()) {
            channel.close().addListener(f -> {
                if (f.isSuccess() && !cf.isCancelled()) {
                    cf.complete(true);
                } else {
                    cf.completeExceptionally(f.cause());
                }
            });
        } else {
            log.info("[{} close] 节点已关闭 channel", this.getClass().getSimpleName(), channel);
        }
        return cf;
    }

    /**
     * 节点摘除
     */
    @Override
    public CompletableFuture<Boolean> kill() {
        setStatus(NodeStatusEnums.KILL);
        return close();
    }

    @Override
    public void revive() {
        status = NodeStatusEnums.NEW;
    }

    @Override
    public int hashCode() {
        return this.key().hashCode();
    }

    @Override
    public boolean equals(Object obj) {
        if (!(obj instanceof INode)) {return false;}
        return this.key().equals(((INode)obj).key());
    }
    
    @Override
	public String toString() {
		return String.format("Node{type:%s, group:%s, key:%s state:%s}", type(), group, key(), status);
	}
	/**
     * 生成新id
     */
    protected long id() {
        return idGenerator.id();
    }


    @Override
    public String group() {
        return group;
    }

    @Override
    public NodeKey key() {
        return nodeKey;
    }

    @Override
    public NodeStatusEnums status() {
        return status;
    }

}
